package io.cici.server.core.instance;

import io.cici.common.enums.TaskInstanceStatus;
import io.cici.common.enums.TimeExpressionType;
import io.cici.common.model.LifeCycle;
import io.cici.common.request.ServerStopTaskInstanceRequest;
import io.cici.common.request.TaskTrackerReportInstanceStatusRequest;
import io.cici.common.utils.CommonUtils;
import io.cici.remote.framework.base.URL;
import io.cici.server.common.module.WorkerInfo;
import io.cici.server.common.utils.ApplicationUtils;
import io.cici.server.core.hide.service.UserService;
import io.cici.server.core.hide.util.alarm.AlarmCenter;
import io.cici.server.core.hide.util.alarm.AlarmUtils;
import io.cici.server.core.hide.util.alarm.module.JobInstanceAlarm;
import io.cici.server.persistence.remote.model.TaskDto;
import io.cici.server.persistence.remote.model.TaskInstanceDto;
import io.cici.server.persistence.remote.model.UserInfoDO;
import io.cici.server.persistence.remote.repository.TaskInstanceRepository;
import io.cici.server.remote.aware.TransportServiceAware;
import io.cici.server.remote.transporter.TransportService;
import io.cici.server.remote.transporter.impl.ServerURLFactory;
import io.cici.server.remote.worker.WorkerClusterQueryService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutionException;

/**
 * 管理被调度的任务实例（状态更新相关）
 *
 * @author tjq
 * @since 2020/4/7
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class TaskInstanceManager implements TransportServiceAware {

    private final AlarmCenter alarmCenter;


    private final TaskMetaDataService taskMetaDataService;

    private final TaskInstanceRepository taskInstanceRepository;


    private final WorkerClusterQueryService workerClusterQueryService;

    /**
     * 基础组件通过 aware 注入，避免循环依赖
     */
    private TransportService transportService;

    /**
     * 更新任务状态
     * ********************************************
     * 2021-02-03 modify by Echo009
     * 实例的执行次数统一在这里管理，对于非固定频率的任务
     * 当 db 中实例的状态为等待派发时，runningTimes + 1
     * ********************************************
     *
     * @param request TaskTracker上报任务实例状态的请求
     */
    public void updateStatus(TaskTrackerReportInstanceStatusRequest request) throws ExecutionException {

        Long taskInstanceId = request.getTaskInstanceId();
        // 获取相关数据
        TaskDto taskDto = taskMetaDataService.fetchJobInfoByInstanceId(request.getTaskInstanceId());
        TaskInstanceDto taskInstanceDto = taskInstanceRepository.findByInstanceId(taskInstanceId);
        if (taskInstanceDto == null) {
            log.warn("[InstanceManager-{}] can't find InstanceInfo from database", taskInstanceId);
            return;
        }

        // 考虑极端情况：Processor 处理耗时小于 server 写 DB 耗时，会导致状态上报时无 taskTracker 地址，此处等待后重新从DB获取数据 GitHub#620
        if (StringUtils.isEmpty(taskInstanceDto.getTaskTrackerAddress())) {
            log.warn("[InstanceManager-{}] TaskTrackerAddress is empty, server will wait then acquire again!", taskInstanceId);
            CommonUtils.easySleep(277);
            taskInstanceDto = taskInstanceRepository.findByInstanceId(taskInstanceId);
        }

        int originStatus = taskInstanceDto.getStatus();
        // 丢弃过期的上报数据
        if (request.getReportTime() <= taskInstanceDto.getLastReportTime()) {
            log.warn("[InstanceManager-{}] receive the expired status report request: {}, this report will be dropped.", taskInstanceId, request);
            return;
        }
        // 丢弃非目标 TaskTracker 的上报数据（脑裂情况）
        if (!request.getSourceAddress().equals(taskInstanceDto.getTaskTrackerAddress())) {
            log.warn("[InstanceManager-{}] receive the other TaskTracker's report: {}, but current TaskTracker is {}, this report will be dropped.", taskInstanceId, request, taskInstanceDto.getTaskTrackerAddress());
            return;
        }

        TaskInstanceStatus receivedTaskInstanceStatus = TaskInstanceStatus.of(request.getInstanceStatus());
        Integer timeExpressionType = taskDto.getTimeExpressionType();
        // 更新 最后上报时间 和 修改时间
        taskInstanceDto.setLastReportTime(request.getReportTime());
        taskInstanceDto.setGmtModified(new Date());

        // FREQUENT 任务没有失败重试机制，TaskTracker一直运行即可，只需要将存活信息同步到DB即可
        // FREQUENT 任务的 newStatus 只有2中情况，一种是 RUNNING，一种是 FAILED（表示该机器 overload，需要重新选一台机器执行）
        // 综上，直接把 status 和 runningNum 同步到DB即可
        if (TimeExpressionType.FREQUENT_TYPES.contains(timeExpressionType)) {
            // 如果实例处于失败状态，则说明该 worker 失联了一段时间，被 server 判定为宕机，而此时该秒级任务有可能已经重新派发了，故需要 Kill 掉该实例
            // fix issue 375
            if (taskInstanceDto.getStatus() == TaskInstanceStatus.FAILED.getCode()) {
                log.warn("[InstanceManager-{}] receive TaskTracker's report: {}, but current instance is already failed, this instance should be killed.", taskInstanceId, request);
                stopInstance(taskInstanceId, taskInstanceDto);//任务失败了怎么处理 需要做失败异常处理  需要单独的处理逻辑。不能喝一般的任务融在一起。失败了需要记录失败机器。如果所有的机器都失败了，然后尝试和报警。
                return;
            }
            LifeCycle lifeCycle = LifeCycle.parse(taskDto.getLifecycle());
            // 检查生命周期是否已结束
            if (lifeCycle.getEnd() != null && lifeCycle.getEnd() <= System.currentTimeMillis()) {
                stopInstance(taskInstanceId, taskInstanceDto);
                taskInstanceDto.setStatus(TaskInstanceStatus.SUCCEED.getCode());
            } else {
                taskInstanceDto.setStatus(receivedTaskInstanceStatus.getCode());
            }
            taskInstanceDto.setResult(request.getResult());
            taskInstanceDto.setRunningTimes(request.getTotalTaskNum());
            taskInstanceRepository.saveAndFlush(taskInstanceDto);
            // 任务需要告警
            if (request.isNeedAlert()) {
                log.info("[InstanceManager-{}] receive frequent task alert req,time:{},content:{}", taskInstanceId, request.getReportTime(), request.getAlertContent());
                alert(taskInstanceId, request.getAlertContent());
            }
            return;
        }
        // 更新运行次数
        if (taskInstanceDto.getStatus() == TaskInstanceStatus.WAITING_WORKER_RECEIVE.getCode()) {
            // 这里不会存在并发问题
            taskInstanceDto.setRunningTimes(taskInstanceDto.getRunningTimes() + 1);
        }
        // QAQ ，不能提前变更 status，否则会导致更新运行次数的逻辑不生效继而导致普通任务 无限重试
        taskInstanceDto.setStatus(receivedTaskInstanceStatus.getCode());

        boolean finished = false;
        if (receivedTaskInstanceStatus == TaskInstanceStatus.SUCCEED) {
            taskInstanceDto.setResult(request.getResult());
            taskInstanceDto.setFinishedTime(request.getEndTime() == null ? System.currentTimeMillis() : request.getEndTime());
            finished = true;
        } else if (receivedTaskInstanceStatus == TaskInstanceStatus.FAILED) {

            // 当前重试次数 <= 最大重试次数，进行重试 （第一次运行，runningTimes为1，重试一次，instanceRetryNum也为1，故需要 =）
            if (taskInstanceDto.getRunningTimes() <= taskDto.getInstanceRetryNum()) {

                log.info("[InstanceManager-{}] instance execute failed but will take the {}th retry.", taskInstanceId, taskInstanceDto.getRunningTimes());

                // 延迟10S重试（由于重试不改变 instanceId，如果派发到同一台机器，上一个 TaskTracker 还处于资源释放阶段，无法创建新的TaskTracker，任务失败）
                taskInstanceDto.setExpectedTriggerTime(System.currentTimeMillis() + 10000);

                // 修改状态为 等待派发，正式开始重试
                // 问题：会丢失以往的调度记录（actualTriggerTime什么的都会被覆盖）
                taskInstanceDto.setStatus(TaskInstanceStatus.WAITING_DISPATCH.getCode());
            } else {
                taskInstanceDto.setResult(request.getResult());
                taskInstanceDto.setFinishedTime(request.getEndTime() == null ? System.currentTimeMillis() : request.getEndTime());
                finished = true;
                log.info("[InstanceManager-{}] instance execute failed and have no chance to retry.", taskInstanceId);
            }
        }
        if (finished) {
            // 最终状态允许直接覆盖更新
            taskInstanceRepository.saveAndFlush(taskInstanceDto);
            // 这里的 InstanceStatus 只有 成功/失败 两种，手动停止不会由 TaskTracker 上报
            processFinishedInstance(taskInstanceId, receivedTaskInstanceStatus, request.getResult());
            return;
        }
        // 带条件更新
        final int i = taskInstanceRepository.updateStatusChangeInfoByInstanceIdAndStatus(taskInstanceDto.getLastReportTime(), taskInstanceDto.getGmtModified(), taskInstanceDto.getRunningTimes(), taskInstanceDto.getStatus(), taskInstanceDto.getInstanceId(), originStatus);
        if (i == 0) {
            log.warn("[InstanceManager-{}] update instance status failed, maybe the instance status has been changed by other thread. discard this status change,{}", taskInstanceId, taskInstanceDto);
        }
    }

    private void stopInstance(Long instanceId, TaskInstanceDto instanceInfo) {
        Optional<WorkerInfo> workerInfoOpt = workerClusterQueryService.getWorkerInfoByAddress(instanceInfo.getAppId(), instanceInfo.getTaskTrackerAddress());
        if (workerInfoOpt.isPresent()) {
            ServerStopTaskInstanceRequest stopInstanceReq = new ServerStopTaskInstanceRequest(instanceId);
            WorkerInfo workerInfo = workerInfoOpt.get();
            final URL url = ServerURLFactory.stopInstance2Worker(workerInfo.getAddress());
            transportService.tell(workerInfo.getProtocol(), url, stopInstanceReq);
        }
    }

    /**
     * 收尾完成的任务实例
     *
     * @param instanceId 任务实例ID
     * @param status     任务状态，有 成功/失败/手动停止
     * @param result     执行结果
     */
    public void processFinishedInstance(Long instanceId, TaskInstanceStatus status, String result) {

        log.info("[Instance-{}] process finished, final status is {}.", instanceId, status.name());

        // 上报日志数据
        //HashedWheelTimerHolder.INACCURATE_TIMER.schedule(() -> instanceLogService.sync(instanceId), 60, TimeUnit.SECONDS);


        // 告警
        if (status == TaskInstanceStatus.FAILED) {
            alert(instanceId, result);
        }
        // 主动移除缓存，减小内存占用
        taskMetaDataService.invalidateJobInfo(instanceId);
    }

    private void alert(Long instanceId, String alertContent) {
        TaskInstanceDto instanceInfo = taskInstanceRepository.findByInstanceId(instanceId);
        TaskDto jobInfo;
        try {
            jobInfo = taskMetaDataService.fetchJobInfoByInstanceId(instanceId);
        } catch (Exception e) {
            log.warn("[InstanceManager-{}] can't find jobInfo, alarm failed.", instanceId);
            return;
        }
        JobInstanceAlarm content = new JobInstanceAlarm();
        BeanUtils.copyProperties(jobInfo, content);
        BeanUtils.copyProperties(instanceInfo, content);
        List<UserInfoDO> userList = ApplicationUtils.getBean(UserService.class).fetchNotifyUserList(jobInfo.getNotifyUserIds());
        if (!StringUtils.isEmpty(alertContent)) {
            content.setResult(alertContent);
        }
        alarmCenter.alarmFailed(content, AlarmUtils.convertUserInfoList2AlarmTargetList(userList));
    }

    @Override
    public void setTransportService(TransportService transportService) {
        this.transportService = transportService;
    }
}
